//package com.shujia.kafka
//
//import com.alibaba.fastjson.{JSON, JSONObject}
//import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
//
//import java.util.Properties
//import scala.io.Source
//
//object Demo2Sendbatch {
//  def main(args: Array[String]): Unit = {
//
//    /**
//     * 1、建立链接
//     */
//    val properties = new Properties()
//    //指定kafka集群列表
//    properties.setProperty("bootstrap.servers", "master:9092,node1:9092,node2:9092")
//
//    //设置key 和value的序列化类
//    properties.setProperty("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
//    properties.setProperty("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")
//
//    val producer = new KafkaProducer[String, String](properties)
//
//    val studentList: List[String] = Source.fromFile("data/students.json").getLines().toList
//
//    //循环发送多条数据
//    for (student <- studentList) {
//      //当topic不存在是会自动创建一个分区为1副本为1的topic
//      //kafka-topics.sh --create --zookeeper master:2181,node1:2181,node2:2181/kafka --replication-factor 3 --partitions 3 --topic student_partition
//      //生产者自己决定向哪一个分区中写数据，默认是轮询
//      //val record = new ProducerRecord[String, String]("student_partition", student)
//
//      /**
//       * hash 分区
//       * 如果数据分布不均会出现数据倾斜
//       */
//      //将相同性别的数据写入同一个分区
//      val jsonObj: JSONObject = JSON.parseObject(student)
//      val gender: String = jsonObj.getString("gender")
//      //对topic的分区数取余
//      val partition: Int = Math.abs(gender.hashCode) % 3
//      val record = new ProducerRecord[String, String]("student_partition", partition, null, student)
//
//      //发送数据到kafka集群
//      producer.send(record)
//    }
//
//    producer.close()
//  }
//
//}
